// Copyright 2024 Zenoh P2P Project
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.

use libp2p::{
    request_response::{self, OutboundRequestId, ProtocolSupport},
    swarm::NetworkBehaviour,
    PeerId,
};
use std::time::Duration;
use futures::{AsyncReadExt, AsyncWriteExt};

use super::protocol::{HandshakeRequest, HandshakeResponse, PROTOCOL_NAME};

/// Configuration for the handshake protocol behaviour
#[derive(Debug, Clone)]
pub struct Config {
    /// Timeout for handshake requests
    pub request_timeout: Duration,
    /// Maximum concurrent handshake requests
    pub max_concurrent_requests: usize,
    /// Whether to support inbound handshake requests
    pub support_inbound: bool,
    /// Whether to support outbound handshake requests
    pub support_outbound: bool,
}

impl Default for Config {
    fn default() -> Self {
        Self {
            request_timeout: Duration::from_secs(30),
            max_concurrent_requests: 10,
            support_inbound: true,
            support_outbound: true,
        }
    }
}

/// Events emitted by the handshake protocol behaviour
#[derive(Debug)]
pub enum Event {
    /// Handshake request received from a peer
    RequestReceived {
        /// Peer ID of the requesting node
        peer_id: PeerId,
        /// Handshake request data
        request: HandshakeRequest,
        /// Channel to send response back
        response_channel: request_response::ResponseChannel<HandshakeResponse>,
    },
    /// Handshake response received from a peer
    ResponseReceived {
        /// Peer ID of the responding node
        peer_id: PeerId,
        /// Handshake response data
        response: HandshakeResponse,
        /// Request ID that this response corresponds to
        request_id: OutboundRequestId,
    },
    /// Handshake request failed (timeout or error)
    RequestFailed {
        /// Peer ID of the target node
        peer_id: PeerId,
        /// Request ID that failed
        request_id: OutboundRequestId,
        /// Error that occurred
        error: super::protocol::HandshakeError,
    },
}

/// Handshake protocol codec implementation
#[derive(Debug, Clone, Default)]
pub struct HandshakeCodec;

#[async_trait::async_trait]
impl request_response::Codec for HandshakeCodec {
    type Protocol = libp2p::StreamProtocol;
    type Request = HandshakeRequest;
    type Response = HandshakeResponse;

    async fn read_request<T>(
        &mut self,
        _protocol: &Self::Protocol,
        io: &mut T,
    ) -> std::io::Result<Self::Request>
    where
        T: futures::AsyncRead + Unpin + Send,
    {
        let mut vec = Vec::new();
        io.read_to_end(&mut vec).await?;
        
        serde_json::from_slice(&vec)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
    }

    async fn read_response<T>(
        &mut self,
        _protocol: &Self::Protocol,
        io: &mut T,
    ) -> std::io::Result<Self::Response>
    where
        T: futures::AsyncRead + Unpin + Send,
    {
        let mut vec = Vec::new();
        io.read_to_end(&mut vec).await?;
        
        serde_json::from_slice(&vec)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
    }

    async fn write_request<T>(
        &mut self,
        _protocol: &Self::Protocol,
        io: &mut T,
        request: Self::Request,
    ) -> std::io::Result<()>
    where
        T: futures::AsyncWrite + Unpin + Send,
    {
        let bytes = serde_json::to_vec(&request)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        
        io.write_all(&bytes).await?;
        Ok(())
    }

    async fn write_response<T>(
        &mut self,
        _protocol: &Self::Protocol,
        io: &mut T,
        response: Self::Response,
    ) -> std::io::Result<()>
    where
        T: futures::AsyncWrite + Unpin + Send,
    {
        let bytes = serde_json::to_vec(&response)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        
        io.write_all(&bytes).await?;
        Ok(())
    }
}

/// Handshake protocol behaviour implementation
#[derive(NetworkBehaviour)]
pub struct Behaviour {
    /// Request-response behaviour for handshake protocol
    request_response: request_response::Behaviour<HandshakeCodec>,
}

impl Behaviour {
    /// Create a new handshake behaviour with default configuration
    pub fn new() -> Self {
        Self::with_config(Config::default())
    }

    /// Create a new handshake behaviour with custom configuration
    pub fn with_config(config: Config) -> Self {
        let protocol_support = match (config.support_inbound, config.support_outbound) {
            (true, true) => ProtocolSupport::Full,
            (true, false) => ProtocolSupport::Inbound,
            (false, true) => ProtocolSupport::Outbound,
            (false, false) => ProtocolSupport::Inbound, // 默认支持inbound
        };

        let request_response_config = request_response::Config::default()
            .with_request_timeout(config.request_timeout)
            .with_max_concurrent_streams(config.max_concurrent_requests);

        let request_response = request_response::Behaviour::with_codec(
            HandshakeCodec,
            [(PROTOCOL_NAME, protocol_support)],
            request_response_config,
        );

        Self {
            request_response,
        }
    }

    /// Send a handshake request to a peer
    pub fn send_request(&mut self, peer_id: PeerId, request: HandshakeRequest) -> OutboundRequestId {
        self.request_response.send_request(&peer_id, request)
    }

    /// Send a handshake response to a peer
    pub fn send_response(
        &mut self,
        response_channel: request_response::ResponseChannel<HandshakeResponse>,
        response: HandshakeResponse,
    ) -> Result<(), super::protocol::HandshakeError> {
        self.request_response
            .send_response(response_channel, response)
            .map_err(|_| super::protocol::HandshakeError::InternalError)
    }
}

impl Default for Behaviour {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_behaviour_creation() {
        let behaviour = Behaviour::new();
        // Behaviour should be successfully created without panicking
        // We can't access internal config fields, so we just verify the behaviour was created
        assert!(std::mem::size_of_val(&behaviour) > 0);
    }

    #[test]
    fn test_config_default() {
        let config = Config::default();
        assert_eq!(config.request_timeout, Duration::from_secs(30));
        assert_eq!(config.max_concurrent_requests, 10);
        assert!(config.support_inbound);
        assert!(config.support_outbound);
    }
}